package kafka

import (
	"context"
	"fmt"
	"time"

	"github.com/segmentio/kafka-go"
)

// Kafka
const (
	kafkaBroker = "localhost:9092"
)

// 生产者逻辑
func ProduceMessage(topic, message string) error {
	writer := kafka.Writer{
		Addr:         kafka.TCP(kafkaBroker),
		Topic:        topic,
		Balancer:     &kafka.LeastBytes{},
		RequiredAcks: kafka.RequireAll,
	}

	defer writer.Close()

	err := writer.WriteMessages(context.Background(),
		kafka.Message{
			Key:   []byte(fmt.Sprintf("key-%d", time.Now().UnixNano())),
			Value: []byte(message),
		},
	)

	return err
}

//
//// Gin 路由处理函数
//func produceHandler(c *gin.Context) {
//	message := c.PostForm("message")
//	if message == "" {
//		c.JSON(http.StatusBadRequest, gin.H{"error": "Message is required"})
//		return
//	}
//
//	err := produceMessage(kafkaBroker, kafkaTopic, message)
//	if err != nil {
//		c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to send message"})
//		return
//	}
//
//	c.JSON(http.StatusOK, gin.H{"status": "Message sent successfully"})
//}
